package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public final class BiAnTickerApplication extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

	//private static final String BA_URL = "wss://stream.binance.com:9443/ws"; //BTCUSDT@bookTicker
	private static final String BA_URL = "wss://stream.binance.com:9443/stream";
//	private static final String BA_URL = "wss://stream.binance.com:9443/ws";

	private static final String METHOD = "method";

	private static final String SUBSCRIBE = "SUBSCRIBE";

	private static final String UNSUBSCRIBE = "UNSUBSCRIBE";

	private static final String PARAMS = "params";

	private static final String ID = "id";

	private static final String BTCUSDT_SUB_VALUE = "@ticker";

	/**
	 * 心跳调度器
	 */

	private long period = 3000;

	@Autowired
	private SymbolService symbolService;

	@Override
	public void afterPropertiesSet() throws Exception {
		super.afterPropertiesSet(new BiAnWebSocketClient(new URI(BA_URL), new Draft_6455(), null, 5000));
		scheduled.scheduleWithFixedDelay(() -> heartBeat(), 30000, period, TimeUnit.MILLISECONDS);
	}

	@Override
	public void destroy() throws Exception {
		this.client.close();
		scheduled.shutdown();
	}

	@Override
	public void subscribe(Symbol symbol) {
		//订阅事件
		JSONArray array = new JSONArray();
		array.add(BourseEnum.BINANCE.getSymbolFormat().symbolFormat(symbol)+BTCUSDT_SUB_VALUE);
		sendMessage(array, SUBSCRIBE);
	}

	@Override
	public void unSubscribe(Symbol symbol) {
		//取消订阅
		JSONArray array = new JSONArray();
		array.add(BourseEnum.BINANCE.getSymbolFormat().symbolFormat(symbol)+BTCUSDT_SUB_VALUE);
		sendMessage(array, UNSUBSCRIBE);
		cleanSymbolPrice(symbol.getName());
	}

	@Override
	public void subscribeList(Set<Symbol> subSet) {
		for (Symbol symbol : subSet) {
			subscribe(symbol);
		}
	}

	@Override
	public BoursePrice getSymbolPrice (String symbol) {
		return symbol_map.get(symbol);
	}

	@Override
	public void cleanSymbolPrice(String symbol) {
		symbol_map.remove(symbol);
	}

	private void sendMessage(JSONArray paramArray, String method) {
		JSONObject json = new JSONObject();
		json.put(METHOD, method);
		json.put(PARAMS, paramArray);
		json.put(ID, sequence.nextId());
		log.error("发送币安的消息：{}", json.toJSONString());
		this.client.send(json.toJSONString());
	}


	private class BiAnWebSocketClient extends WebSocketClient {

		public BiAnWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders,
				int connectTimeout) {
			super(serverUri, protocolDraft, httpHeaders, connectTimeout);
		}

		@Override
		public void onOpen(ServerHandshake handshakedata) {
			log.info("币安WebSocket接口: {} 链接成功.", this.getURI());
			Set<Symbol> subSet = symbolService.listByBourse(BourseEnum.BINANCE.getValue());
			if (subSet != null && !subSet.isEmpty()) {
				//Set<String> topicSet = subSet.stream().map(item ->  BourseEnum.BINANCE.getSymbolFormat().symbolFormat(item)+ BTCUSDT_SUB_VALUE).collect(Collectors.toSet());
				log.info("币安订阅开始{}",subSet);
				subscribeList(subSet);
			}
		}

		@Override
		public void onMessage(ByteBuffer bytes) {
			byte[] array = bytes.array();
			// array = HuoBiGZipUtils.decompress(array);
			String str = new String(array, StandardCharsets.UTF_8);
			log.info("收到币安消息：{}", str);
			if(str.contains(PING)) {
				this.send(str.replace(PING, PONG));
			}else if(str.contains("tick")){
				JSONObject object = JSON.parseObject(str);
				JSONObject tick = object.getJSONObject("tick");
				log.info("收到币安的消息：{}",tick);
			}else {
				log.error("币安WebSocket数据: {} 未做处理", str);
			}
		}

		@Override
		public void onMessage(String s) {
			JSONObject res = JSONObject.parseObject(s);
			if (res != null && res.getString("stream") != null && res.getString("stream").contains(BTCUSDT_SUB_VALUE)) {
				String symbol =res.getString("stream").replace(BTCUSDT_SUB_VALUE,"");
				JSONObject data = res.getJSONObject("data");
				data.put("updateTime",new Date());
				BoursePrice price = BoursePrice.builder().buyPrice(data.getBigDecimal("b")).buyNum(data.getBigDecimal("B")).sellPrice(data.getBigDecimal("a")).sellNum(data.getBigDecimal("A")).data(data.toJSONString()).updateTime(new Date()).build();
				//  log.info("币安获取到【***{}***】的实时价格为：{}", symbol, price.toString());
				symbol_map.put(symbol, price);
//				log.info("币安{}获取买一价格为{}，卖一价格为{}", symbol, data.getString("b"), data.getString("a"));
			} else {
				log.info("获取到币安信息，未作处理：{}",res);
			}
		}

		@Override
		public void onClose(int code, String reason, boolean remote) {
			log.info("币安Websocket接口已关闭, 原因: {}, 状态: {}", reason, code);
		}

		@Override
		public void onError(Exception e) {
			log.error("币安WebSocket接口，发生异常", e);
		}

	}
}
